package cn._51doit.live.udf;

import cn._51doit.live.pojo.DataBean;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava18.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class IsNewUserFunctionV2 extends KeyedProcessFunction<String, DataBean, DataBean> {

    private transient ValueState<BloomFilter<String>> bloomFilterState;

    @Override
    public void open(Configuration parameters) throws Exception {
        //定义状态描述器
        ValueStateDescriptor<BloomFilter<String>> stateDescriptor = new ValueStateDescriptor<>("uid-bloom-filter-state", TypeInformation.of(new TypeHint<BloomFilter<String>>() {}));
        bloomFilterState = getRuntimeContext().getState(stateDescriptor);
    }

    @Override
    public void processElement(DataBean bean, Context ctx, Collector<DataBean> out) throws Exception {

        String deviceId = bean.getDeviceId();

        BloomFilter<String> bloomFilter = bloomFilterState.value();
        if (bloomFilter == null) {
            bloomFilter = BloomFilter.create(Funnels.unencodedCharsFunnel(), 100000);
        }
        if(!bloomFilter.mightContain(deviceId)) {
            bloomFilter.put(deviceId);
            bean.setIsN(1); //设置当前用户为新用户
        }
        bloomFilterState.update(bloomFilter);

        out.collect(bean);

    }
}
